package com.sfx.messagequeue.mqServer.datacenter;

import com.sfx.messagequeue.common.exception.MQException;
import com.sfx.messagequeue.common.utils.BinaryPool;
import com.sfx.messagequeue.mqServer.core.MSGQueue;
import com.sfx.messagequeue.mqServer.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

import static com.sfx.messagequeue.common.constants.MessageConstants.*;

/**
 * Created with IntelliJ IDEA.
 * Description: 主要针对磁盘上的文件进行管理
 * User: sfx
 * Date: 2023-08-01
 * Time: 16:56
 */
public class MessageFileManager {

    public static class Stat {
        //总的消息数量
        public int totalCount;
        //有效消息数量
        public int validCount;
    }

    public void init() {
    }


    /**
     * 获取磁盘上的队列路径
     */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /**
     * 获取磁盘上队列的queue_data文件路径
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /**
     * 获取磁盘上队列的queue_stat文件路径
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * 获取新文件路径
     */
    private String getDataQueueNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new_file.txt";
    }

    /**
     * 读取stat
     */
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (FileInputStream fileInputStream = new FileInputStream(getQueueStatPath(queueName))) {
            //使用Scanner去读
            Scanner scanner = new Scanner(fileInputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("读取Stat失败!!!");
        }
        return null;
    }

    /**
     * 写入stat
     */
    private void writeStat(String queueName, Stat stat) {
        try (FileOutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            //使用Scanner去写
            PrintWriter writer = new PrintWriter(outputStream);
            writer.write(stat.totalCount + "\t" + stat.validCount);
            //将缓冲区内容刷新到磁盘
            writer.flush();
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("写入stat失败!!!");
        }
    }

    /**
     * 创建队列所在的目录和文件
     */
    public void createQueueFiles(String queueName) throws IOException {
        //创建队列所在目录
        File queueDir = new File(getQueueDir(queueName));
        if (!queueDir.exists()) {
            //不存在则创建
            boolean ok = queueDir.mkdirs();
            if (!ok) {
                throw new RuntimeException("创建queueDir文件失败 " + queueDir.getAbsolutePath());
            }
        }

        //创建消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new RuntimeException("创建queue_data文件失败 " + queueDataFile.getAbsolutePath());
            }
        }

        //创建消息统计文件
        File queueStatFile = new File(getQueueDataPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new RuntimeException("创建queue_stat文件失败 " + queueStatFile.getAbsolutePath());
            }
        }

        //初始化文件统计信息
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName, stat);
    }

    /**
     * 删除队列所在的目录和文件
     */
    public boolean deleteFiles(String queueName) {
        //删除目录
        File queueDir = new File(getQueueDir(queueName));
        boolean ok1 = queueDir.delete();

        //删除queue_data文件
        File queueDataPath = new File(getQueueDataPath(queueName));
        boolean ok2 = queueDataPath.delete();

        //删除queue_stat文件
        File queueStatPath = new File(getQueueStatPath(queueName));
        boolean ok3 = queueStatPath.delete();

        File queueDataNewPath = new File(getDataQueueNewPath(queueName));
        boolean ok4 = queueDataPath.delete();

        return ok1 && ok2 && ok3 && ok4;
    }

    /**
     * 查询队列数据文件和统计文件是否都存在
     * 后续生产者给Broker生产消息,如果消息持久化,就需要写入到文件中,就先保证消息存储文件是否存在
     */
    private boolean checkFileExists(String queueName) {
        //queue_data文件是否存在
        File queueDataPath = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataPath.exists();

        //queue_stat文件是否存在
        File queueStatPath = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatPath.exists();
        return ok1 && ok2;
    }

    /**
     * 通过这个方法我们将消息存储到文件中
     */
    public void sendMessage(MSGQueue queue, Message message) throws MQException, IOException {
        //1.检查该队列是否存在
        if (!checkFileExists(queue.getName())) {
            throw new MQException("[MessageFileManger] : 队列文件不存在,无法存储消息到队列中");
        }
        //2.解决线程安全问题的说明
        //2.1queue对象的问题
        //此处传的queue对象一定是内存中的queue,不会为其他对象,所以不会存在对不同的对象加锁
        //所以这里就是对同一个queue对象加锁
        //2.2 加锁的问题
        //这里加上synchronized的原因是,当多个线程去写文件的时候会出现线程安全问题
        //比如 线程A计算出偏移量,刚要去写文件,线程B就穿插进来计算偏移量,写文件,等线程B写文件执行完成之后
        //线程A继续完成写文件,但是此时线程A计算出的偏移量的位置已经被线程B写入一个消息了
        //如果线程A继续写入消息,那么消息就会与内存中的偏移量不一一对应,导致消息损坏了
        synchronized (queue) {
            File messageQueueFile = new File(getQueueDataPath(queue.getName()));
            //2.将Message对象序列化为字节数组,以便传输,存储到文件中
            byte[] msgBinaryByteArray = BinaryPool.toByteArray(message);
            //3.我们将Message对象存储到文件之前,我们要先把offsetBeg,offsetEnd先计算出来.
            //我们根据之前的消息格式约定,进行计算
            //offsetBeg : 文件长度 + 4(存储消息长度4字节)
            //offsetEnd : 文件长度 + 消息长度 + 4(存储消息长度4字节)
            long offsetBeg = messageQueueFile.length() + MESSAGE_LENGTH_FORM;
            long offsetEnd = messageQueueFile.length() + msgBinaryByteArray.length + MESSAGE_LENGTH_FORM;
            message.setOffsetBeg(offsetBeg);
            message.setOffsetEnd(offsetEnd);
            //4.写入到文件中,写的时候一定是追加写,也就是追加到文件的末尾,而不是覆盖原先文件,加一个参数true即可
            try (FileOutputStream outputStream = new FileOutputStream(messageQueueFile, true);) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //DataOutputStream的write方法可以一次往文件中写入4个字节
                    //按照之前的规定,先写入消息的长度,在写入消息主题
                    //4.1写入消息长度
                    dataOutputStream.writeInt(msgBinaryByteArray.length);
                    //4.1写入消息主体
                    dataOutputStream.write(msgBinaryByteArray);
                }
            }
            //5.写入文件之后,我们还要做一下统计,将统计信息写入到文件中
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 通过这个方法把Message在文件中进行逻辑删除
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException, MQException {
        //1.检查该队列是否存在
        if (!checkFileExists(queue.getName())) {
            throw new MQException("[MessageFileManger] : 队列文件不存在,无法存储消息到队列中");
        }
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw");) {
                //2.将消息从文件中读取出来
                //2.1读入到byte数组中
                byte[] messageBytes = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                //2.2移动光标到offsetBeg
                randomAccessFile.seek(message.getOffsetBeg());
                //2.3读满字节数组
                randomAccessFile.read(messageBytes);
                //2.4将字节数组反序列化为Java Bean
                Message diskMessage = (Message) BinaryPool.toJavaBean(messageBytes);

                //3.把Message对象标记为无效状态
                diskMessage.setIsValid((byte) 0x0);

                //4.将消息重新写入到文件中
                //4.1从新定位光标到offsetBeg
                randomAccessFile.seek(message.getOffsetBeg());
                //4.2将diskMessage序列化文字节数组
                byte[] diskMessageBytes = BinaryPool.toByteArray(diskMessage);
                //4.3写入到文件中
                randomAccessFile.write(diskMessageBytes);
                //5.写入文件之后,我们还要做一下统计,将统计信息写入到文件中
                Stat stat = readStat(queue.getName());
                if (stat.validCount > 0) {
                    stat.totalCount -= 1;
                }
                writeStat(queue.getName(), stat);
            }
        }
    }

    /**
     * 将文件中的消息加载到内存中,放入一个链表中存储
     * 这个方法在程序启动的时候就开始执行
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MQException, ClassNotFoundException {
        //1.检查该队列是否存在
        if (!checkFileExists(queueName)) {
            throw new MQException("[MessageFileManger] : 队列文件不存在,无法存储消息到队列中");
        }
        //2.读取消息到链表中
        LinkedList<Message> messages = new LinkedList<>();
        try (FileInputStream inputStream = new FileInputStream(getQueueDataPath(queueName));) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                long currentOffset = 0;//读取文件的光标,刚开始从文件头按照消息的格式进行读取.
                while (true) {
                    //如果读取文件结尾,他会抛出EOFException异常
                    //2.1读取消息的长度
                    int messageSize = dataInputStream.readInt();
                    //2.2读取消息本体(Message对象)
                    byte[] messageBytes = new byte[messageSize];
                    int actualSize = dataInputStream.read(messageBytes);
                    if (actualSize != messageSize) {
                        //消息有损坏,实际消息长度与规定格式长度不匹配
                        throw new MQException("消息有损坏,实际消息长度与规定格式长度不匹配!!!");
                    }
                    //2.3将读取的字节数组数据序列化为Message对象
                    Message message = (Message) BinaryPool.toJavaBean(messageBytes);
                    if (message.getIsValid() != 0x1) {
                        //无效的消息
                        currentOffset += (MESSAGE_LENGTH_FORM + messageSize);
                        continue;
                    }
                    //2.4设置消息的offset
                    message.setOffsetBeg(currentOffset + MESSAGE_LENGTH_FORM);
                    message.setOffsetEnd(currentOffset + MESSAGE_LENGTH_FORM + messageSize);
                    //2.5更新文件光标
                    currentOffset += (MESSAGE_LENGTH_FORM + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                System.out.println("[MessageFileManger] : 读取消息完成");
            }
        }
        return messages;
    }

    /**
     * 检查是否要垃圾回收
     * 如果总消息数量>2000 && 有效数量/总数量 < 0.5 就进行垃圾回收
     */
    public boolean checkGC(String queueName) {
        //1.得到文件消息的统计信息
        Stat stat = readStat(queueName);
        //2.判定条件
        return stat.totalCount >= MESSAGE_GC_MAX_LIMIT &&
                (double) stat.validCount / (double) stat.totalCount < MESSAGE_GC_MAX_RATIO;
    }

    /**
     * 执行GC
     * 基本步骤
     * 1.将旧文件的有效消息读取出来
     * 2.将有效消息写入到新的文件中
     * 3.将新文件重命名为旧文件名,删除旧文件
     * 4.更新消息的统计信息
     */
    public void gc(MSGQueue queue) throws IOException, MQException, ClassNotFoundException {
        synchronized (queue) {
            long beg = System.currentTimeMillis();
            //1.检查文件是否存在
            if (!checkFileExists(queue.getName())) {
                throw new MQException("[MessageFileManger] : 队列文件不存在,无法存储消息到队列中");
            }

            //2.读取旧文件的有效消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            //3.创建新的文件
            File dataQueueFile = new File(getDataQueueNewPath(queue.getName()));
            if (dataQueueFile.exists()) {
                throw new MQException("[MessageFileManger] : 文件已经存在");
            }

            //不存在才创建新的文件
            boolean ok1 = dataQueueFile.createNewFile();
            if (!ok1) {
                throw new MQException("[MessageFileManger] : 创建文件失败");
            }

            //4.将旧文件消息写入到新的文件中
            try (FileOutputStream outputStream = new FileOutputStream(dataQueueFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] messageBytes = BinaryPool.toByteArray(message);
                        //写入消息长度
                        dataOutputStream.writeInt(messageBytes.length);
                        //写入消息本体
                        dataOutputStream.write(messageBytes);
                    }
                }
            }

            //5.删除旧的文件
            File oldFile = new File(getQueueDataPath(queue.getName()));
            boolean ok3 = oldFile.delete();
            if (!ok3) {
                throw new MQException("MessageFileManger] :删除旧文件失败  queueOldFile=" + oldFile.getAbsolutePath());
            }

            //6.新文件名重命名为旧的文件名
            boolean ok2 = dataQueueFile.renameTo(oldFile);
            if (!ok2) {
                throw new MQException("MessageFileManger] :新文件重命名失败 queueOldFile=" + oldFile.getAbsolutePath()
                + "queueNewFile=" + dataQueueFile.getAbsolutePath());
            }

            //7.更新统计信息
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            long end = System.currentTimeMillis();
            System.out.println("[MessageFileManger] " + "queueName=" + queue.getName() +" gc执行完毕 消耗时长为 : " + (end-beg) + "ms");
        }
    }
}